package com.wande.dataplatform.etl.node;

import cn.hutool.core.util.StrUtil;
import com.wande.dataplatform.common.utils.DataSourceUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 输出节点 - 将数据写入目标数据源
 *
 * @author dataplatform
 */
@Slf4j
@Component
public class OutputNode implements EtlNode {

    @Override
    public EtlNodeResult execute(EtlContext context) {
        try {
            List<Map<String, Object>> data = context.getDataSet();
            if (data == null || data.isEmpty()) {
                log.info("输出节点: 没有数据需要写入");
                return EtlNodeResult.success(data);
            }
            
            Map<String, Object> config = context.getConfig();
            
            // 获取配置
            String jdbcUrl = (String) config.get("jdbcUrl");
            String username = (String) config.get("username");
            String password = (String) config.get("password");
            String tableName = (String) config.get("tableName");
            String writeMode = (String) config.getOrDefault("writeMode", "insert");
            
            log.info("输出节点开始写入数据: {} 行到表 {}", data.size(), tableName);
            
            long successCount = 0;
            long failedCount = 0;
            
            try (Connection conn = DataSourceUtils.getConnection(jdbcUrl, username, password)) {
                conn.setAutoCommit(false);
                
                for (Map<String, Object> row : data) {
                    try {
                        String sql = buildInsertSql(tableName, row, writeMode);
                        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                            int paramIndex = 1;
                            for (Object value : row.values()) {
                                stmt.setObject(paramIndex++, value);
                            }
                            
                            if ("replace".equals(writeMode)) {
                                // 对于replace模式，需要设置两次参数（INSERT和UPDATE部分）
                                for (Object value : row.values()) {
                                    stmt.setObject(paramIndex++, value);
                                }
                            }
                            
                            stmt.executeUpdate();
                            successCount++;
                        }
                    } catch (Exception e) {
                        log.warn("写入行数据失败: {}", e.getMessage());
                        failedCount++;
                    }
                }
                
                conn.commit();
            }
            
            log.info("输出节点写入完成: 成功 {} 行, 失败 {} 行", successCount, failedCount);
            context.getStatistics().addTargetRows(successCount);
            
            EtlNodeResult result = EtlNodeResult.success(data);
            result.setSuccessRows(successCount);
            result.setFailedRows(failedCount);
            return result;
            
        } catch (Exception e) {
            log.error("输出节点执行失败", e);
            return EtlNodeResult.failure("输出节点执行失败: " + e.getMessage());
        }
    }

    /**
     * 构建INSERT SQL
     */
    private String buildInsertSql(String tableName, Map<String, Object> row, String writeMode) {
        List<String> columns = row.keySet().stream().collect(Collectors.toList());
        String columnStr = String.join(", ", columns);
        String placeholders = columns.stream().map(c -> "?").collect(Collectors.joining(", "));
        
        if ("replace".equals(writeMode)) {
            // INSERT ... ON DUPLICATE KEY UPDATE
            String updatePart = columns.stream()
                .map(col -> col + " = ?")
                .collect(Collectors.joining(", "));
            return String.format("INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
                tableName, columnStr, placeholders, updatePart);
        } else {
            // 普通INSERT
            return String.format("INSERT INTO %s (%s) VALUES (%s)",
                tableName, columnStr, placeholders);
        }
    }

    @Override
    public String getNodeType() {
        return OUTPUT;
    }

    @Override
    public boolean validate(Map<String, Object> config) {
        return config.containsKey("jdbcUrl")
            && config.containsKey("username")
            && config.containsKey("tableName");
    }
}
